package com.alarm.ecp;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

public class CepDemo {

    public static void main(String[] args) throws Exception {
        // 创建流式计算上下文环境
        StreamExecutionEnvironment  env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 生成DataStream
        DataStream<String> dataStream = env.fromElements("hello world");
        // 生成模式（规则）（pattern对象）
        /**
         * 个体匹配模式
         * 1、单例模式：匹配一个事件
         * 2、循环模式：匹配多个事件或1个事件，单例模式+量词（times）
         */
        Pattern<String, ?> pattern = Pattern.<String>begin("login").where(new SimpleCondition<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                return false;
            }
        }).times(3);

        /**
         * 组合模式
         * 1、next: 严格紧邻（连续）
         * 2、followedBy: 宽松紧邻（非连续）
         * 3、followedByAny: 非严格匹配，比followedBy更宽松
         */
        Pattern<String, ?> pattern2 = Pattern.<String>begin("login")//.where()
                .followedBy("sale"); // .where()

        // 将pattern对象与数据源关联
        PatternStream<String> patternStream = CEP.pattern(dataStream, pattern);

        // 通过PatternStream对象的select（）方法，将符合规则的数据提取输出
        SingleOutputStreamOperator<String> select = patternStream.process(new PatternProcessFunction<String, String>() {
            @Override
            public void processMatch(Map<String, List<String>> map, Context context, Collector<String> collector) throws Exception {
                collector.collect("");
            }
        });
    }
}
